﻿/**************************************************************
 * Copyright Eastmoney Corporation. All rights reserved.
 * 
 * Author: guitao(guitao@eastmoney.com) 
 * Create Date: 2020/6/15 14:57:53
 * Description:
 *          
 * Revision History:
 *      Date         Author               Description
 *              
***************************************************************/

using Confluent.Kafka;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;

namespace gt.EventBus.Kafka
{
    public class DefaultKafkaPersistentConnection : IKafkaPersistentConnection
    {
        private readonly IProducer<string, string> _producer;
        private IConsumer<string, string> _consumer;

        private readonly ConsumerConfig _consumerConfig;
        private readonly ProducerConfig _producerConfig;
        private readonly string _topicName;

        public DefaultKafkaPersistentConnection(ProducerConfig producerConfig, ConsumerConfig consumerConfig, string topicName)
        {
            _producerConfig = producerConfig ?? throw new ArgumentNullException("producerConfig");
            _consumerConfig = consumerConfig ?? throw new ArgumentNullException("consumerConfig");

            _producer = new ProducerBuilder<string, string>(_producerConfig).Build();
            _consumer = new ConsumerBuilder<string, string>(_consumerConfig)
                                .SetErrorHandler((_, e) => Console.WriteLine($"error:{e.Reason}")).Build();
            _consumer.Subscribe(topicName);
            _topicName = topicName;
        }

        public string TopicName { get { return _topicName; } }

        public IConsumer<string, string> Consumer { get { return _consumer; } }

        public IProducer<string, string> Producer { get { return _producer; } }

        public void Dispose()
        {
            if (_producer != null)
            {
                try
                {
                    _producer.Flush();
                    _producer.Dispose();
                }
                catch { }
            }
            if (_consumer != null)
            {
                try
                {
                    _consumer.Close();
                }
                catch { }
            }
        }
    }
}
